package com.etlmaster.datax.plugins.writer.rationaldbwriter;

import java.sql.Connection;
import java.util.Properties;

import org.apache.commons.lang.StringUtils;

import com.dp.nebula.wormhole.common.AbstractPlugin;
import com.dp.nebula.wormhole.common.JobStatus;
import com.dp.nebula.wormhole.common.WormholeException;
import com.dp.nebula.wormhole.common.interfaces.IParam;
import com.dp.nebula.wormhole.common.interfaces.ISourceCounter;
import com.dp.nebula.wormhole.common.interfaces.ITargetCounter;
import com.dp.nebula.wormhole.common.interfaces.IWriterPeriphery;
import com.dp.nebula.wormhole.plugins.common.DBSource;
import com.dp.nebula.wormhole.plugins.common.DBUtils;
import com.etlmaster.executor.utils.LogWriter;

public class RationalDbWriterPeriphery implements IWriterPeriphery{
	
	private Connection conn;
	
	private String username;
	
	private String password;

	private String url;
	
	private String encode;
	
	private String rollback;
	
	private String pre;
	
	private String post;
	
	private int concurrency = 1;
	
	private String countSql;
	
	private String writerID;
	
	private String driver;
	
	private String validationSql;
	
	private int errorCodeAdd;
	
	public void prepare(IParam param, ISourceCounter counter) {
		register(param);
		if(pre.isEmpty()){
			return;
		}
		try{
			conn = DBSource.getConnection(this.getClass(), "", "", url);
			String[] sqlArray = pre.split(";");
			for(String sql:sqlArray){
				sql = sql.trim();
				if(sql.isEmpty()) {
					continue;
				}
				DBUtils.update(conn, sql);
			}
			conn.close();
		}catch (Exception e) {
			LogWriter.addLog("ERROR",writerID + ": RationalDbWriter prepare failed. ");
			throw new WormholeException(e,JobStatus.PRE_WRITE_FAILED.getStatus() + errorCodeAdd,writerID);
		}
	}
	
	public void rollback(IParam param) {
		if(rollback.isEmpty()){
			LogWriter.addLog("ERROR","Fail to roll back because rollback sql is empty!");
			return;
		}
		try{
			conn = DBSource.getConnection(this.getClass(), "", "", url);
			String[] sqlArray = rollback.split(";");
			for(String sql:sqlArray){
				sql = sql.trim();
				if(sql.isEmpty()) {
					continue;
				}
				DBUtils.update(conn, sql);
			}
			conn.close();
		} catch (Exception e) {
			LogWriter.addLog("ERROR",writerID + ": RationalDbWriter roll back failed. ");
			throw new WormholeException(e,JobStatus.ROLL_BACK_FAILED.getStatus() + errorCodeAdd,writerID);
		}
	}

	
	
	private void init(IParam param){
		/* for database connection */
		this.username 	  = param.getValue(ParamKey.USER_NAME,"");
		this.password 	  = param.getValue(ParamKey.PASSWORD,"");
	
		this.url		  = param.getValue(ParamKey.URL,"");
		this.encode       = param.getValue(ParamKey.ENCODING, "");
		this.rollback     = param.getValue(ParamKey.ROLL_BACK,"");
		this.pre          = param.getValue(ParamKey.PRE,"");
		this.post         = param.getValue(ParamKey.POST,"");
		this.countSql     = param.getValue(ParamKey.COUNT_SQL,"");
		this.writerID	  = param.getValue(AbstractPlugin.PLUGINID, "");
		int priority      = param.getIntValue(ParamKey.PRIORITY, 0);
		this.driver       = param.getValue(ParamKey.driverClass, "");
		this.validationSql= param.getValue(ParamKey.validationSql, "");
		this.errorCodeAdd = RationalDbWriter.PLUGIN_NO*JobStatus.PLUGIN_BASE + priority*JobStatus.WRITER_BASE;	}
	
	private void register(IParam param) {
		init(param);
		/* for connection session */
		Properties p = createProperties();
		try {
			DBSource.register(RationalDbWriter.class, "", "", this.url, p);;
		} catch (Exception e) {
			throw new WormholeException(e, JobStatus.WRITE_CONNECTION_FAILED.getStatus() + errorCodeAdd);
		}
		
	}
		
	private Properties createProperties() {
		Properties p = new Properties();
		
		String encodeDetail = "";
		
		if(!StringUtils.isBlank(this.encode)){
			encodeDetail = "useUnicode=true&characterEncoding="	+ this.encode + "&";
		}
		String url = this.url;
		
		p.setProperty("driverClassName", this.driver);
		p.setProperty("url", url);
		p.setProperty("username", username);
		p.setProperty("password", password);
		p.setProperty("maxActive", String.valueOf(concurrency + 2));
		//don't set initialSize, otherwise accessToUnderlyingConnectionAllowed will not be set successfully
		//p.setProperty("initialSize", String.valueOf(concurrency + 2));
		p.setProperty("maxIdle", "1");
		p.setProperty("maxWait", "1000");
		p.setProperty("defaultReadOnly", "false");
		p.setProperty("testOnBorrow", "true");
		p.setProperty("validationQuery", this.validationSql);

		LogWriter.addLog("DEBUG",String.format("Rational Db try connection: %s .", url));
		return p;
	}

	public void doPost(IParam param, ITargetCounter counter) {
		
		
	}
}